// PassengerPatternReducer.java
package com.taxi.analysis.passengerTrip;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * 乘客行程模式分析Reducer
 * 功能：分析每个乘客的行程模式，计算复购周期等指标
 * 技术：二次排序数据处理 + 模式分析
 */
public class PassengerPatternReducer extends Reducer<PassengerTripKey, Text, Text, Text> {

    private SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Override
    public void reduce(PassengerTripKey key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        String passengerId = key.getPassengerId();

        // 存储该乘客的所有行程（已按时间排序）
        List<TripRecord> trips = new ArrayList<>();

        // 解析所有行程记录
        for (Text value : values) {
            try {
                TripRecord trip = parseTripRecord(value.toString());
                if (trip != null) {
                    trips.add(trip);
                }
            } catch (Exception e) {
                System.err.println("解析行程记录错误: " + value.toString());
            }
        }

        // 如果该乘客有有效行程数据，进行模式分析
        if (!trips.isEmpty()) {
            analyzePassengerPattern(passengerId, trips, context);
        }
    }

    /**
     * 分析乘客行程模式
     */
    private void analyzePassengerPattern(String passengerId, List<TripRecord> trips,
                                         Context context) throws IOException, InterruptedException {

        int totalTrips = trips.size();
        if (totalTrips < 2) {
            // 单次行程乘客，输出基本信息
            outputSingleTripPassenger(passengerId, trips.get(0), context);
            return;
        }

        // 1. 计算基础统计指标
        PatternAnalysis analysis = calculateBasicStats(trips);

        // 2. 分析时间模式
        analyzeTimePatterns(analysis, trips);

        // 3. 分析地理模式
        analyzeGeographicPatterns(analysis, trips);

        // 4. 分析消费模式
        analyzeSpendingPatterns(analysis, trips);

        // 5. 计算复购周期
        calculateRepurchaseCycle(analysis, trips);

        // 6. 识别行程链
        identifyTripChains(analysis, trips);

        // 7. 输出分析结果
        outputPatternAnalysis(passengerId, analysis, context);
    }

    /**
     * 计算基础统计指标
     */
    private PatternAnalysis calculateBasicStats(List<TripRecord> trips) {
        PatternAnalysis analysis = new PatternAnalysis();

        analysis.totalTrips = trips.size();
        analysis.firstTripTime = trips.get(0).pickupTime;
        analysis.lastTripTime = trips.get(trips.size() - 1).pickupTime;

        // 计算总体时间跨度（天）
        long timeSpanMs = analysis.lastTripTime.getTime() - analysis.firstTripTime.getTime();
        analysis.timeSpanDays = timeSpanMs / (24 * 60 * 60 * 1000);

        // 计算平均指标
        double totalDistance = 0, totalAmount = 0, totalDuration = 0;
        for (TripRecord trip : trips) {
            totalDistance += trip.tripDistance;
            totalAmount += trip.totalAmount;
            totalDuration += trip.tripDuration;
        }

        analysis.avgDistance = totalDistance / trips.size();
        analysis.avgAmount = totalAmount / trips.size();
        analysis.avgDuration = totalDuration / trips.size();
        analysis.totalSpending = totalAmount;

        return analysis;
    }

    /**
     * 分析时间模式
     */
    private void analyzeTimePatterns(PatternAnalysis analysis, List<TripRecord> trips) {
        Map<String, Integer> timePeriodCounts = new HashMap<>();
        List<Long> intervals = new ArrayList<>();

        for (int i = 0; i < trips.size(); i++) {
            TripRecord trip = trips.get(i);

            // 统计时间段偏好
            timePeriodCounts.put(trip.timePeriod,
                    timePeriodCounts.getOrDefault(trip.timePeriod, 0) + 1);

            // 计算行程间隔
            if (i > 0) {
                long interval = trip.pickupTime.getTime() - trips.get(i-1).dropoffTime.getTime();
                intervals.add(interval / (60 * 1000)); // 转换为分钟
            }
        }

        // 找出最偏好的时间段
        analysis.preferredTimePeriod = findMostFrequent(timePeriodCounts);

        // 计算平均行程间隔
        if (!intervals.isEmpty()) {
            analysis.avgIntervalMinutes = intervals.stream()
                    .mapToLong(Long::longValue)
                    .average()
                    .orElse(0.0);
        }

        // 分析行程频率模式
        analysis.tripFrequencyPattern = analyzeTripFrequency(analysis.totalTrips, analysis.timeSpanDays);
    }

    /**
     * 分析地理模式
     */
    private void analyzeGeographicPatterns(PatternAnalysis analysis, List<TripRecord> trips) {
        Map<String, Integer> pickupZoneCounts = new HashMap<>();
        Map<String, Integer> dropoffZoneCounts = new HashMap<>();
        Map<String, Integer> routeCounts = new HashMap<>();

        for (TripRecord trip : trips) {
            // 统计上车地点偏好
            pickupZoneCounts.put(trip.pickupZone,
                    pickupZoneCounts.getOrDefault(trip.pickupZone, 0) + 1);

            // 统计下车地点偏好
            dropoffZoneCounts.put(trip.dropoffZone,
                    dropoffZoneCounts.getOrDefault(trip.dropoffZone, 0) + 1);

            // 统计路线偏好
            String route = trip.pickupZone + " -> " + trip.dropoffZone;
            routeCounts.put(route, routeCounts.getOrDefault(route, 0) + 1);
        }

        analysis.preferredPickupZone = findMostFrequent(pickupZoneCounts);
        analysis.preferredDropoffZone = findMostFrequent(dropoffZoneCounts);
        analysis.preferredRoute = findMostFrequent(routeCounts);

        // 计算地理多样性（访问的不同地点数量）
        analysis.pickupZoneDiversity = pickupZoneCounts.size();
        analysis.dropoffZoneDiversity = dropoffZoneCounts.size();
    }

    /**
     * 分析消费模式
     */
    /**
     * 分析消费模式
     */
    private void analyzeSpendingPatterns(PatternAnalysis analysis, List<TripRecord> trips) {
        Map<String, Integer> fareLevelCounts = new HashMap<>();
        Map<String, Integer> tripTypeCounts = new HashMap<>();

        for (TripRecord trip : trips) {
            // 统计费用等级偏好
            fareLevelCounts.put(trip.fareLevel,
                    fareLevelCounts.getOrDefault(trip.fareLevel, 0) + 1);

            // 统计行程类型偏好
            tripTypeCounts.put(trip.tripType,
                    tripTypeCounts.getOrDefault(trip.tripType, 0) + 1);
        }

        analysis.preferredFareLevel = findMostFrequent(fareLevelCounts);
        analysis.preferredTripType = findMostFrequent(tripTypeCounts);

        // 计算消费稳定性（标准差）
        double avgAmount = analysis.avgAmount;
        double variance = trips.stream()
                .mapToDouble(trip -> Math.pow(trip.totalAmount - avgAmount, 2))
                .average()
                .orElse(0.0);
        analysis.spendingStability = Math.sqrt(variance);
    }

    /**
     * 计算复购周期
     */
    private void calculateRepurchaseCycle(PatternAnalysis analysis, List<TripRecord> trips) {
        if (trips.size() < 2) {
            analysis.repurchaseCycleDays = 0.0;
            return;
        }

        List<Double> cycleDays = new ArrayList<>();

        for (int i = 1; i < trips.size(); i++) {
            long intervalMs = trips.get(i).pickupTime.getTime() - trips.get(i-1).pickupTime.getTime();
            double intervalDays = intervalMs / (24.0 * 60 * 60 * 1000);
            cycleDays.add(intervalDays);
        }

        // 计算平均复购周期
        analysis.repurchaseCycleDays = cycleDays.stream()
                .mapToDouble(Double::doubleValue)
                .average()
                .orElse(0.0);

        // 计算复购周期稳定性
        double avgCycle = analysis.repurchaseCycleDays;
        double cycleVariance = cycleDays.stream()
                .mapToDouble(cycle -> Math.pow(cycle - avgCycle, 2))
                .average()
                .orElse(0.0);
        analysis.repurchaseCycleStability = Math.sqrt(cycleVariance);

        // 分类复购模式
        analysis.repurchasePattern = classifyRepurchasePattern(analysis.repurchaseCycleDays,
                analysis.repurchaseCycleStability);
    }

    /**
     * 识别行程链（连续行程）
     */
    private void identifyTripChains(PatternAnalysis analysis, List<TripRecord> trips) {
        List<TripChain> chains = new ArrayList<>();
        TripChain currentChain = null;

        for (int i = 0; i < trips.size(); i++) {
            TripRecord trip = trips.get(i);

            // 判断是否为连续行程（间隔小于2小时）
            boolean isContinuous = trip.isContinuousTrip;

            if (isContinuous && currentChain != null) {
                // 添加到当前行程链
                currentChain.addTrip(trip);
            } else {
                // 开始新的行程链
                if (currentChain != null && currentChain.getTripCount() > 1) {
                    chains.add(currentChain);
                }
                currentChain = new TripChain();
                currentChain.addTrip(trip);
            }
        }

        // 添加最后一个行程链
        if (currentChain != null && currentChain.getTripCount() > 1) {
            chains.add(currentChain);
        }

        analysis.tripChains = chains;
        analysis.totalChains = chains.size();
        analysis.avgChainLength = chains.isEmpty() ? 0 :
                chains.stream().mapToInt(TripChain::getTripCount).average().orElse(0.0);
    }

    /**
     * 输出单次行程乘客信息
     */
    private void outputSingleTripPassenger(String passengerId, TripRecord trip, Context context)
            throws IOException, InterruptedException {

        String result = String.format("单次乘客|%s|%s|%s|%.2f|%.2f|%s|%s|%s",
                trip.pickupZone, trip.dropoffZone, trip.timePeriod,
                trip.tripDistance, trip.totalAmount, trip.tripType,
                trip.fareLevel, trip.passengerType);

        context.write(new Text("单次行程_" + passengerId), new Text(result));
    }

    /**
     * 输出完整的模式分析结果
     */
    private void outputPatternAnalysis(String passengerId, PatternAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        // 1. 输出基础统计信息
        String basicStats = String.format("基础统计|总行程:%d|时间跨度:%.1f天|平均距离:%.2f|平均费用:%.2f|总消费:%.2f|平均时长:%.2f分钟",
                analysis.totalTrips, analysis.timeSpanDays, analysis.avgDistance,
                analysis.avgAmount, analysis.totalSpending, analysis.avgDuration);

        context.write(new Text("基础统计_" + passengerId), new Text(basicStats));

        // 2. 输出时间模式分析
        String timePattern = String.format("时间模式|偏好时段:%s|平均间隔:%.1f分钟|频率模式:%s",
                analysis.preferredTimePeriod, analysis.avgIntervalMinutes,
                analysis.tripFrequencyPattern);

        context.write(new Text("时间模式_" + passengerId), new Text(timePattern));

        // 3. 输出地理模式分析
        String geoPattern = String.format("地理模式|偏好上车地:%s|偏好下车地:%s|偏好路线:%s|上车地多样性:%d|下车地多样性:%d",
                analysis.preferredPickupZone, analysis.preferredDropoffZone,
                analysis.preferredRoute, analysis.pickupZoneDiversity,
                analysis.dropoffZoneDiversity);

        context.write(new Text("地理模式_" + passengerId), new Text(geoPattern));

        // 4. 输出消费模式分析
        String spendingPattern = String.format("消费模式|偏好费用等级:%s|偏好行程类型:%s|消费稳定性:%.2f",
                analysis.preferredFareLevel, analysis.preferredTripType,
                analysis.spendingStability);

        context.write(new Text("消费模式_" + passengerId), new Text(spendingPattern));

        // 5. 输出复购周期分析
        String repurchaseAnalysis = String.format("复购周期|平均周期:%.2f天|周期稳定性:%.2f|复购模式:%s",
                analysis.repurchaseCycleDays, analysis.repurchaseCycleStability,
                analysis.repurchasePattern);

        context.write(new Text("复购周期_" + passengerId), new Text(repurchaseAnalysis));

        // 6. 输出行程链分析
        if (analysis.totalChains > 0) {
            String chainAnalysis = String.format("行程链|总链数:%d|平均链长:%.1f|最长链:%d行程",
                    analysis.totalChains, analysis.avgChainLength,
                    analysis.tripChains.stream().mapToInt(TripChain::getTripCount).max().orElse(0));

            context.write(new Text("行程链_" + passengerId), new Text(chainAnalysis));

            // 输出详细的行程链信息
            for (int i = 0; i < Math.min(3, analysis.tripChains.size()); i++) {
                TripChain chain = analysis.tripChains.get(i);
                String chainDetail = String.format("行程链详情|链%d|行程数:%d|总距离:%.2f|总费用:%.2f|持续时间:%.1f小时",
                        i+1, chain.getTripCount(), chain.getTotalDistance(),
                        chain.getTotalAmount(), chain.getDurationHours());

                context.write(new Text("行程链详情_" + passengerId + "_" + (i+1)), new Text(chainDetail));
            }
        }

        // 7. 输出综合评估
        String overallAssessment = generateOverallAssessment(analysis);
        context.write(new Text("综合评估_" + passengerId), new Text(overallAssessment));
    }

    /**
     * 生成综合评估
     */
    private String generateOverallAssessment(PatternAnalysis analysis) {
        StringBuilder assessment = new StringBuilder("综合评估|");

        // 乘客价值评估
        String valueLevel = assessPassengerValue(analysis);
        assessment.append("价值等级:").append(valueLevel).append("|");

        // 忠诚度评估
        String loyaltyLevel = assessLoyalty(analysis);
        assessment.append("忠诚度:").append(loyaltyLevel).append("|");

        // 行为稳定性评估
        String stabilityLevel = assessStability(analysis);
        assessment.append("行为稳定性:").append(stabilityLevel).append("|");

        // 推荐策略
        String recommendation = generateRecommendation(analysis);
        assessment.append("推荐策略:").append(recommendation);

        return assessment.toString();
    }

    /**
     * 评估乘客价值
     */
    private String assessPassengerValue(PatternAnalysis analysis) {
        double monthlySpending = analysis.totalSpending / (analysis.timeSpanDays / 30.0);

        if (monthlySpending >= 200) {
            return "高价值";
        } else if (monthlySpending >= 100) {
            return "中等价值";
        } else if (monthlySpending >= 50) {
            return "一般价值";
        } else {
            return "低价值";
        }
    }

    /**
     * 评估忠诚度
     */
    private String assessLoyalty(PatternAnalysis analysis) {
        double tripsPerWeek = analysis.totalTrips / (analysis.timeSpanDays / 7.0);

        if (tripsPerWeek >= 5) {
            return "高忠诚";
        } else if (tripsPerWeek >= 2) {
            return "中等忠诚";
        } else if (tripsPerWeek >= 0.5) {
            return "一般忠诚";
        } else {
            return "低忠诚";
        }
    }

    /**
     * 评估行为稳定性
     */
    private String assessStability(PatternAnalysis analysis) {
        // 基于复购周期稳定性和消费稳定性
        double cycleCV = analysis.repurchaseCycleStability / analysis.repurchaseCycleDays;
        double spendingCV = analysis.spendingStability / analysis.avgAmount;

        double overallStability = (cycleCV + spendingCV) / 2;

        if (overallStability <= 0.3) {
            return "非常稳定";
        } else if (overallStability <= 0.5) {
            return "较稳定";
        } else if (overallStability <= 0.8) {
            return "一般稳定";
        } else {
            return "不稳定";
        }
    }

    /**
     * 生成推荐策略
     */
    private String generateRecommendation(PatternAnalysis analysis) {
        if (analysis.totalTrips >= 10 && analysis.avgAmount >= 30) {
            return "VIP服务推荐";
        } else if (analysis.repurchaseCycleDays <= 7) {
            return "频次奖励计划";
        } else if (analysis.totalChains > 0) {
            return "连程优惠推荐";
        } else if (analysis.preferredTimePeriod.contains("高峰")) {
            return "高峰时段优先服务";
        } else {
            return "基础服务维护";
        }
    }

    /**
     * 分类复购模式
     */
    private String classifyRepurchasePattern(double avgCycle, double stability) {
        if (avgCycle <= 1) {
            return "日常高频";
        } else if (avgCycle <= 7) {
            if (stability <= 2) {
                return "周期规律";
            } else {
                return "周期不规律";
            }
        } else if (avgCycle <= 30) {
            return "月度使用";
        } else {
            return "偶尔使用";
        }
    }

    /**
     * 分析行程频率模式
     */
    private String analyzeTripFrequency(int totalTrips, double timeSpanDays) {
        double tripsPerDay = totalTrips / timeSpanDays;

        if (tripsPerDay >= 2) {
            return "高频用户";
        } else if (tripsPerDay >= 0.5) {
            return "中频用户";
        } else if (tripsPerDay >= 0.1) {
            return "低频用户";
        } else {
            return "偶尔用户";
        }
    }

    /**
     * 找出最频繁的项目
     */
    private String findMostFrequent(Map<String, Integer> counts) {
        return counts.entrySet().stream()
                .max(Map.Entry.comparingByValue())
                .map(entry -> entry.getKey() + "(" + entry.getValue() + "次)")
                .orElse("未知");
    }

    /**
     * 解析行程记录
     */
    private TripRecord parseTripRecord(String recordString) {
        try {
            String[] fields = recordString.split("\\|");
            if (fields.length >= 16) {
                TripRecord record = new TripRecord();

                record.tripSequence = Integer.parseInt(fields[0]);
                record.pickupTime = dateFormat.parse(fields[1]);
                record.dropoffTime = dateFormat.parse(fields[2]);
                record.pickupZone = fields[3];
                record.dropoffZone = fields[4];
                record.tripDistance = Double.parseDouble(fields[5]);
                record.totalAmount = Double.parseDouble(fields[6]);
                record.tripDuration = Double.parseDouble(fields[7]);
                record.timeGapMinutes = Double.parseDouble(fields[8]);
                record.timePeriod = fields[9];
                record.tripType = fields[10];
                record.fareLevel = fields[11];
                record.passengerType = fields[12];
                record.totalTrips = Integer.parseInt(fields[13]);
                record.avgSpending = Double.parseDouble(fields[14]);
                record.isContinuousTrip = "是".equals(fields[15]);

                return record;
            }
        } catch (Exception e) {
            System.err.println("解析行程记录错误: " + recordString);
        }
        return null;
    }

    /**
     * 行程记录内部类
     */
    private static class TripRecord {
        int tripSequence;
        Date pickupTime;
        Date dropoffTime;
        String pickupZone;
        String dropoffZone;
        double tripDistance;
        double totalAmount;
        double tripDuration;
        double timeGapMinutes;
        String timePeriod;
        String tripType;
        String fareLevel;
        String passengerType;
        int totalTrips;
        double avgSpending;
        boolean isContinuousTrip;
    }

    /**
     * 模式分析结果内部类
     */
    private static class PatternAnalysis {
        int totalTrips;
        Date firstTripTime;
        Date lastTripTime;
        double timeSpanDays;
        double avgDistance;
        double avgAmount;
        double avgDuration;
        double totalSpending;
        String preferredTimePeriod;
        double avgIntervalMinutes;
        String tripFrequencyPattern;
        String preferredPickupZone;
        String preferredDropoffZone;
        String preferredRoute;
        int pickupZoneDiversity;
        int dropoffZoneDiversity;
        String preferredFareLevel;
        String preferredTripType;
        double spendingStability;
        double repurchaseCycleDays;
        double repurchaseCycleStability;
        String repurchasePattern;
        List<TripChain> tripChains;
        int totalChains;
        double avgChainLength;
    }

    /**
     * 行程链内部类
     */
    private static class TripChain {
        private List<TripRecord> trips = new ArrayList<>();

        public void addTrip(TripRecord trip) {
            trips.add(trip);
        }

        public int getTripCount() {
            return trips.size();
        }

        public double getTotalDistance() {
            return trips.stream().mapToDouble(trip -> trip.tripDistance).sum();
        }

        public double getTotalAmount() {
            return trips.stream().mapToDouble(trip -> trip.totalAmount).sum();
        }

        public double getDurationHours() {
            if (trips.isEmpty()) return 0;
            long durationMs = trips.get(trips.size()-1).dropoffTime.getTime() -
                    trips.get(0).pickupTime.getTime();
            return durationMs / (60.0 * 60 * 1000);
        }
    }
}

